import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;

public class Demo2API {
    Configuration conf = null;
    Connection conn = null;

    @Before
    public void init() {
        conf = HBaseConfiguration.create();
        // 其他配置参考http://hbase.apache.org/1.4/book.html#config.files
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        try {
            conn = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 创建表
    @Test
    public void create_table() throws IOException {
        Admin admin = conn.getAdmin();
        // admin 相当于HBase的管理员
        // 创建表 传入表名(TableName.valueOf())
        HTableDescriptor tableName = new HTableDescriptor(TableName.valueOf("tableName"));
        // 创建列簇 传入列簇名
        HColumnDescriptor cf1 = new HColumnDescriptor("cf1");
        // 对列簇进行一些配置
        cf1.setMaxVersions(5); // 设置版本号
        cf1.setTimeToLive(30); // 设置TTL时间
        // 将创建好的列簇加入表
        tableName.addFamily(cf1);
        // 使用admin对象创建表
        admin.createTable(tableName);
    }

    // 删除表
    @Test
    public void drop_table() throws IOException {
        Admin admin = conn.getAdmin();
        String tableName = "tableName";
        admin.disableTable(TableName.valueOf(tableName));
        admin.deleteTable(TableName.valueOf(tableName));
    }

    // 添加一条数据
    @Test
    public void put() throws IOException {
        // 如果想要插入数据 查询数据 需要使用Table对象
        // 如果需要对表进行修改，获取表的一些配置、结构 使用HTableDescriptor对象
        Table testJavaAPI = conn.getTable(TableName.valueOf("testJavaAPI"));

        Put put = new Put("00001".getBytes());
        put.addColumn("cf1".getBytes(), "name".getBytes(), "zhangSan".getBytes());

        testJavaAPI.put(put);


    }

    // 获取一条数据
    @Test
    public void get() throws IOException {
        Table testJavaAPI = conn.getTable(TableName.valueOf("testJavaAPI"));
        Get get = new Get("00001".getBytes());
        Result rs = testJavaAPI.get(get);
        byte[] value = rs.getValue("cf1".getBytes(), "name".getBytes());
        System.out.println(Bytes.toString(value));
    }

    // 指定rowkey范围 扫描表
    @Test
    public void scan() throws IOException {
        Table testJavaAPI = conn.getTable(TableName.valueOf("test1"));
        Scan scan = new Scan();
        // 包含startRow 不包含 endRow
        scan.withStartRow("001".getBytes());
        scan.withStopRow("007".getBytes());
        ResultScanner scanner = testJavaAPI.getScanner(scan);
        Result rs = scanner.next();
        while (rs != null) {

            byte[] row = rs.getRow();// 获取rowkey
            String rk = Bytes.toString(row);
            System.out.println();
            if ("001".equals(rk)) {
                byte[] value = rs.getValue("cf1".getBytes(), "name".getBytes());
                System.out.println(Bytes.toString(value));
            } else if ("002".equals(rk)) {
                byte[] value = rs.getValue("cf1".getBytes(), "name0".getBytes());
                System.out.println(Bytes.toString(value));
                byte[] value1 = rs.getValue("cf1".getBytes(), "name1".getBytes());
                System.out.println(Bytes.toString(value1));
                byte[] value2 = rs.getValue("cf1".getBytes(), "name100".getBytes());
                System.out.println(Bytes.toString(value2));
                byte[] value3 = rs.getValue("cf1".getBytes(), "name2".getBytes());
                System.out.println(Bytes.toString(value3));
                byte[] value4 = rs.getValue("cf1".getBytes(), "name3".getBytes());
                System.out.println(Bytes.toString(value4));
                byte[] value5 = rs.getValue("cf1".getBytes(), "name4".getBytes());
                System.out.println(Bytes.toString(value5));
                byte[] value6 = rs.getValue("cf1".getBytes(), "name5".getBytes());
                System.out.println(Bytes.toString(value6));
            } else if ("007".equals(rk)) {
                byte[] value6 = rs.getValue("cf1".getBytes(), "name".getBytes());
                System.out.println(Bytes.toString(value6));
                byte[] value7 = rs.getValue("cf1".getBytes(), "age1".getBytes());
                System.out.println(Bytes.toString(value7));
            }

            rs = scanner.next();
        }
    }

    @Test
    public void cellUtil() throws IOException {
        Table testJavaAPI = conn.getTable(TableName.valueOf("test1"));
        Scan scan = new Scan();
        // 包含startRow 不包含 endRow
        scan.withStartRow("001".getBytes());
        scan.withStopRow("007".getBytes());
        ResultScanner scanner = testJavaAPI.getScanner(scan);
        Result rs = scanner.next();
        while (rs != null) {
            for (Cell cell : rs.listCells()) {
                byte[] rk = CellUtil.cloneRow(cell);
                byte[] cf = CellUtil.cloneFamily(cell);
                byte[] qualifier = CellUtil.cloneQualifier(cell);
                byte[] value = CellUtil.cloneValue(cell);

                System.out.println("rowkey:" + Bytes.toString(rk) + ",columnsFamily:" + Bytes.toString(cf) + ",qualifier:" + Bytes.toString(qualifier) + ",value:" + Bytes.toString(value));
            }

            rs = scanner.next();
        }
    }

    // 读取文件并写入HBase
    @Test
    public void putAll() throws IOException {
        Admin admin = conn.getAdmin();
        // 判断表是否存在，不存在即创建
        if (!admin.tableExists(TableName.valueOf("students"))) {
            HTableDescriptor students = new HTableDescriptor(TableName.valueOf("students"));
            HColumnDescriptor info = new HColumnDescriptor("info");
            students.addFamily(info);
            admin.createTable(students);
        }

        Table students = conn.getTable(TableName.valueOf("students"));
        // 读取文件
        BufferedReader br = new BufferedReader(new FileReader("data/students.txt"));
        String line = br.readLine();
        while (line != null) {
            String[] splits = line.split(",");
            String rk = splits[0];
            String name = splits[1];
            String age = splits[2];
            String gender = splits[3];
            String clazz = splits[4];

            Put put = new Put(Bytes.toBytes(rk));
            put.addColumn("info".getBytes(), "name".getBytes(), name.getBytes());
            put.addColumn("info".getBytes(), "age".getBytes(), age.getBytes());
            put.addColumn("info".getBytes(), "gender".getBytes(), gender.getBytes());
            put.addColumn("info".getBytes(), "clazz".getBytes(), clazz.getBytes());

            students.put(put);

            line = br.readLine();
        }
        br.close();

    }

    @After
    public void closeAll() throws IOException {
        if (conn != null) {
            conn.close();
        }
    }
}
